#ifndef __M_MSG_H__
#define __M_MSG_H__
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include <iostream>
#include <unordered_map>
#include <mutex>
#include <memory>
#include <list>

namespace castlemq
{
    #define DATAFILE_SUBFIX ".mqd"
    #define TMPILE_SUBFIX ".mqd.tmp"
    using MessagePtr=std::shared_ptr<castlemq::Message>;
    class MessageMapper
    {
        public:
            MessageMapper(std::string& basedir,const std::string& qname):_qname(qname)
            {
                if(basedir.back()!='/') basedir.push_back('/');
                _datafile=basedir+qname+DATAFILE_SUBFIX;
                _tmpfile=basedir+qname+TMPILE_SUBFIX;
                if(!fileHelper(_datafile).exists())
                    assert(fileHelper::createDictory(basedir));
                createMsgFile();
            }
            bool createMsgFile()
            {
                if(fileHelper(_datafile).exists())  return true;
                bool ret=fileHelper::createFile(_datafile);
                if(!ret)
                {
                    DLOG("创建队列文件 %s 失败 ",_datafile.c_str());
                    return false;
                }
                return true;
            }
            void removeMsgFile()
            {
                fileHelper::deleteFile(_datafile);
                fileHelper::deleteFile(_tmpfile);
            }
            bool insert(MessagePtr& msg)
            {
                return insert(_datafile,msg);
            }

           
            bool remove(MessagePtr& msg)
            {
                //1.将消息的有效标志位修改为'0'
                msg->mutable_payload()->set_valid("0");
                //2.对msg进行序列化
                std::string body=msg->payload().SerializeAsString();
                if(body.size()!=msg->length())
                {
                    DLOG("新生成的数据与原数据长度不匹配，不能修改数据");
                    return false;
                }
                //3.将序列化后的消息，写入到数据在文件中的指定位置（覆盖原有数据）
                fileHelper helper(_datafile);
                bool ret=helper.write(body.c_str(),msg->offset(),body.size());
                if(!ret)
                {
                    DLOG("向队列数据文件写入数据失败！");
                    return false;
                }
                return true;
            }

            //加载有效历史消息
            std::list<MessagePtr> gc()
            {
                std::list<MessagePtr> result;
                bool ret;
                //1.加载出文件中有所的有效数据    4字节|数据|4字节|数据...
                ret=load(result);
                if(!ret)
                {
                    DLOG("加载历史数据失败！");
                    return result;
                }
                //2.将有效数据序列化存储到临时文件
                fileHelper::createFile(_tmpfile);
                for(auto& msg:result)
                {
                    ret=insert(_tmpfile,msg);
                    if(!ret)
                    {
                        DLOG("向临时文件写入消息数据失败！");
                        return result;
                    }
                }
                //3.删除源文件
                ret=fileHelper::deleteFile(_datafile);
                if(!ret)
                {
                    DLOG("删除源文件失败！");
                    return result;
                }

                //4.修改临时文件名为源文件名称
                ret=fileHelper(_tmpfile).rename(_datafile);
                if(ret)
                {
                    DLOG("修改临时文件名为源文件名称失败！");
                    return result;
                }
                //5.返回新的有效数据
                return result;
            }
        private:
            bool load(std::list<MessagePtr>& result)
            {
                fileHelper data_file_helper(_datafile);
                size_t offset=0,msg_size;
                size_t fsize=data_file_helper.size();
                bool ret;
                while(offset<fsize)
                {
                    ret=data_file_helper.read((char*)&msg_size,offset,sizeof(size_t));
                    if(!ret)
                    {
                        DLOG("读取消息长度失败！");
                        return false;
                    }
                    offset+=sizeof(size_t);
                    // std::string msg_body(offset,'\0');//你这里长度设置的不对...应该将bodu的大小设置为msg_size
                    std::string msg_body;
                    msg_body.resize(msg_size);
                    data_file_helper.read(&msg_body[0],offset,msg_size);
                    if(!ret)
                    {
                        DLOG("读取消息内容失败！");
                        return false;
                    }
                    offset+=msg_size;

                    MessagePtr msgp=std::make_shared<Message>();
                    msgp->mutable_payload()->ParseFromString(msg_body);
                    
                    if(msgp->payload().valid()=="0") continue;
                    result.push_back(msgp);
                }
                return true;
                
            }
            bool insert(const std::string& filename,MessagePtr& msg)
            {
                //新增数据都是添加文件末尾
                //1.消息序列化
                std::string body=msg->payload().SerializeAsString();
                //2.获取文件长度
                fileHelper helper(filename);
                size_t fsize=helper.size();

                size_t msg_size=body.size();
                //先写入4字节长度，再写入指定长度
                bool ret=helper.write((char*)&msg_size,fsize,sizeof(size_t));
                if(!ret)
                {
                    DLOG("向队列数据文件写入数据长度失败！");
                    return false;
                }
                //3.将数据写入文件指定位置
                ret=helper.write(body.c_str(),fsize+sizeof(size_t),msg_size);
                if(!ret)
                {
                    DLOG("向队列数据文件写入数据内容失败！");
                    return false;
                }
                //4.更新message中的实际存储位置
                msg->set_offset(fsize+sizeof(size_t));
                msg->set_length(body.size());
                return true;
            }
        private:
            std::string _qname;
            std::string _datafile;
            std::string _tmpfile;//临时文件
    };

    class QueueMessage
    {
        public:
            using Ptr=std::shared_ptr<QueueMessage>;
            QueueMessage(std::string& basedir,const std::string& qname):
            _mapper(basedir,qname),_qname(qname),_valid_count(0),_total_count(0)
            {}
            bool recovery()
            {
                //恢复历史消息
                std::unique_lock<std::mutex> lock(_mutex);
                _msgs=_mapper.gc();
                for(auto msg : _msgs)
                {
                    _durable_msgs.insert(std::make_pair(msg->payload().properties().id(),msg));
                }
                _valid_count=_total_count=_msgs.size();
                return true;
            }
            bool insert(const BasicProperties* bp,const std::string& body,bool queue_is_durable)
            {
                //1.构造消息对象
                MessagePtr msg=std::make_shared<Message>();
                msg->mutable_payload()->set_body(body);
                if(bp!=nullptr)
                {
                    DeliveryMode mode = queue_is_durable ? bp->delivery_mode() : DeliveryMode::UNDURABLE;
                    msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                    msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                    msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
                }else{
                    DeliveryMode mode = queue_is_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                    msg->mutable_payload()->mutable_properties()->set_id(UuidHelper::Mkuuid());
                    msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                    msg->mutable_payload()->mutable_properties()->set_routing_key("");
                }

                std::unique_lock<std::mutex> lock(_mutex);
                //2.判断消息是否需要持久化
                if(msg->payload().properties().delivery_mode()== DeliveryMode::DURABLE)
                {
                    msg->mutable_payload()->set_valid("1");//在持久化数据中表示数据有效
                    //3.进行持久化存储
                    if(!_mapper.insert(msg))
                    {
                        DLOG("持久化存储消息 %s 失败！",body.c_str());
                        return false;
                    }
                    _valid_count++;
                    _total_count++;
                    _durable_msgs.insert(std::make_pair(msg->payload().properties().id(),msg));
                }
                //4.内存的管理
                _msgs.push_back(msg);
                return true;
            }
            MessagePtr front()
            {
                std::unique_lock<std::mutex> lock(_mutex);
                if(_msgs.size()==0) return MessagePtr();
                //获取队首消息：从msgs中取出数据
                MessagePtr msg=_msgs.front();
                _msgs.pop_front();
                //将该消息对象，向待确认的hash表中添加一份，等到消息确认后进行删除
                _waitack_msgs.insert(std::make_pair(msg->payload().properties().id(),msg));
                return msg;
            }

            //每次删除消息后，判断是否需要垃圾回收
            bool remove(const std::string& msg_id)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                //1.从待确认队列中查找消息
                auto it = _waitack_msgs.find(msg_id);
                if(it == _waitack_msgs.end())
                {
                    DLOG("没有找到要删除的信息：%s！",msg_id.c_str());
                    return true;
                }
                //2.根据消息持久化模式，决定是否删除持久化信息
                if(it->second->payload().properties().delivery_mode()==DeliveryMode::DURABLE)
                {
                    //3.删除持久化信息
                    _mapper.remove(it->second);
                    _durable_msgs.erase(msg_id);
                    _valid_count--;
                    gc();//内部判断是否需要垃圾回收
                }
                //4.删除内存中信息
                _waitack_msgs.erase(msg_id);
                return true;
            }
            //获取可推送消息数量
            size_t getable_count()
            {
                std::unique_lock<std::mutex> lock(_mutex);
                return _msgs.size();
            }
            size_t total_count()
            {
                std::unique_lock<std::mutex> lock(_mutex);
                return _total_count;
            }
            size_t durable_count()
            {
                std::unique_lock<std::mutex> lock(_mutex);
                return _durable_msgs.size();
            }
            size_t waitack_count()
            {
                std::unique_lock<std::mutex> lock(_mutex);
                return _waitack_msgs.size();
            }
            void clear()
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _mapper.removeMsgFile();
                _msgs.clear();
                _durable_msgs.clear();
                _waitack_msgs.clear();
                _valid_count=0;
                _total_count=0;
            }
            
        private:
            //持久化的消息总量大于2000，并且其中有效比例低于%50
            //则需要进行垃圾清理
            bool GCCheck()
            {
                if(_total_count>2000&&_valid_count*10/_total_count<5)
                {
                    return true;
                }
                return false;
            }

            void gc()
            {
                //1.进行垃圾回收，获取到垃圾回收之后有效的消息信息链表
                if(!GCCheck())  return;
                std::list<MessagePtr> msgs=_mapper.gc();
                for(auto& msg:msgs)
                {
                    auto it=_durable_msgs.find(msg->payload().properties().id());
                    if(it==_durable_msgs.end())
                    {
                        DLOG("垃圾回收后，有一条持久化消息在内存中没有进行管理！");
                        _msgs.push_back(msg);
                        _durable_msgs.insert(std::make_pair(msg->payload().properties().id(),msg));
                        continue;
                    }
                    //2.更新每一条消息的实际存储位置
                    it->second->set_offset(msg->offset());
                    it->second->set_length(msg->length());
                }

                //3.更新当前的有效消息 & 总的持久化消息数量
                _valid_count=_total_count=_msgs.size();
            }
        private:
            std::mutex _mutex;
            std::string _qname;
            size_t _valid_count;
            size_t _total_count;
            MessageMapper _mapper;
            std::list<MessagePtr> _msgs;//待推送消息链表
            std::unordered_map<std::string,MessagePtr> _durable_msgs;//持久化消息hash
            std::unordered_map<std::string,MessagePtr> _waitack_msgs;//待确认消息hash
    };

    class MessageManager
    {
        public:
            using ptr=std::shared_ptr<MessageManager>;
            MessageManager(const std::string& basedir):_basedir(basedir)
            {}
            void initQueueMessage(const std::string& qname)
            {
                QueueMessage::Ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _queue_msgs.find(qname);
                    if(it!=_queue_msgs.end())
                    {
                        return;
                    }
                    qmp=std::make_shared<QueueMessage>(_basedir,qname);
                    _queue_msgs.insert(std::make_pair(qname,qmp));
                }
                qmp->recovery();
            }
            void destroyQueueMessage(const std::string& qname)
            {
                QueueMessage::Ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _queue_msgs.find(qname);
                    if(it==_queue_msgs.end())
                    {
                        return;
                    }
                    qmp=it->second;
                    _queue_msgs.erase(it);
                }
                qmp->clear();
            }
            bool insert(const std::string& qname,BasicProperties* bp,const std::string& body,bool queue_is_durable)
            {
                QueueMessage::Ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _queue_msgs.find(qname);
                    if(it==_queue_msgs.end())
                    {
                        DLOG("向队列 %s 新增消息失败，没有找到消息管理句柄！",qname.c_str());
                        return false;
                    }
                    qmp=it->second;
                }
                return qmp->insert(bp,body,queue_is_durable);
            }
            MessagePtr front(const std::string& qname)
            {
                QueueMessage::Ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _queue_msgs.find(qname);
                    if(it==_queue_msgs.end())
                    {
                        DLOG("获取队列 %s 队首消息失败，没有找到消息管理句柄！",qname.c_str());
                        return MessagePtr();
                    }
                    qmp=it->second;
                }
                return qmp->front();
            }
            void ack(const std::string& qname,const std::string& msg_id)
            {
                QueueMessage::Ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _queue_msgs.find(qname);
                    if(it==_queue_msgs.end())
                    {
                        DLOG("确认队列 %s 消息 %s 失败，没有找到消息管理句柄！",qname.c_str(),msg_id.c_str());
                        return;
                    }
                    qmp=it->second;
                }
                qmp->remove(msg_id);
                return;
            }
            //获取可推送消息数量
            size_t getable_count(const std::string& qname)
            {
                QueueMessage::Ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _queue_msgs.find(qname);
                    if(it==_queue_msgs.end())
                    {
                        DLOG("获取队列 %s 待推送消息数量失败，没有找到消息管理句柄！",qname.c_str());
                        return 0;
                    }
                    qmp=it->second;
                }
                return qmp->getable_count();

            }
            size_t total_count(const std::string& qname)
            {
                QueueMessage::Ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _queue_msgs.find(qname);
                    if(it==_queue_msgs.end())
                    {
                        DLOG("获取队列 %s 总消息消息数量失败，没有找到消息管理句柄！",qname.c_str());
                        return 0;
                    }
                    qmp=it->second;
                }
                return qmp->total_count();
            }
            size_t durable_count(const std::string& qname)
            {
                QueueMessage::Ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _queue_msgs.find(qname);
                    if(it==_queue_msgs.end())
                    {
                        DLOG("获取队列 %s 有效持久化消息数量失败，没有找到消息管理句柄！",qname.c_str());
                        return 0;
                    }
                    qmp=it->second;
                }
                return qmp->durable_count();
            }
            size_t waitack_count(const std::string& qname)
            {
                QueueMessage::Ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _queue_msgs.find(qname);
                    if(it==_queue_msgs.end())
                    {
                        DLOG("获取队列 %s 待确认消息数量失败，没有找到消息管理句柄！",qname.c_str());
                        return 0;
                    }
                    qmp=it->second;
                }
                return qmp->waitack_count();
            }
            void clear()
            {
                std::unique_lock<std::mutex> lock(_mutex);
                for(auto& qmsg:_queue_msgs)
                {
                    qmsg.second->clear();
                }
            }
        private:
            std::mutex _mutex;
            std::string _basedir;
            std::unordered_map<std::string,QueueMessage::Ptr> _queue_msgs;
    };
}
#endif